package com.jscloud.bigdata.flink.flinksql.overwindows;

import com.jscloud.bigdata.flink.tableapi.groupwindow.TempSensorData;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * #基于Process-time排序向前5s开窗
 *   按process-time排序有界向前5s开窗
 *
 * 数据准备如下：
 * s-5,1645085900,14,
 * s-5,1645085901,17,
 * s-5,1645085902,22,
 * s-5,1645085903,7,
 * s-5,1645085904,21,
 * s-5,1645085905,23,
 * s-5,1645085906,8,
 * s-5,1645085907,32,
 * s-5,1645085908,15,
 * s-5,1645085909,9,
 */
public class FlinkSQLOverWinProcessTimeRanger {

        public static void main(String[] args) {
                //1.获取stream的执行环境
                StreamExecutionEnvironment senv= StreamExecutionEnvironment.getExecutionEnvironment();
                senv.setParallelism(1);
                //2.创建表执行环境
                StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv);

                //3.读取数据
                DataStream<TempSensorData> tempSensorData=senv.socketTextStream("bigdata01",8888)
                        .map(event -> {
                                String[] arr = event.split(",");
                                return TempSensorData.builder()
                                        .sensorID(arr[0])
                                        .tp(Long.parseLong(arr[1]))
                                        .temp(Integer.parseInt(arr[2]))
                                        .build();
                        });

                //4.流转换为动态表
                Table table = tEnv.fromDataStream(tempSensorData,
                        $("sensorID"),$("tp"),$("temp"),$("ptTime").proctime());

                //5.自定义窗口并计算
                Table resultTable = tEnv.sqlQuery("select "+
                        "sensorID,"+
                        "max(temp) OVER w AS max_temp,"+
                        "avg(temp) OVER w AS avg_temp "+
                        "from "+table+" WINDOW w AS (\n" +
                        " PARTITION BY sensorID\n" +
                        " ORDER BY ptTime\n" +
                        " RANGE BETWEEN INTERVAL '5' second PRECEDING AND CURRENT ROW)\n");

                //6.执行Flink
                resultTable.execute().print();
        }
}